package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	"bytes"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
)

// import "bytes"
// import "../labgob"

//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in Lab 3 you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh; at that point you can add fields to
// ApplyMsg, but set CommandValid to false for these other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
}

type Entry struct {
	Command interface{}
	Term    int
}

const FOLLOWER int = 1
const CANDIDATE int = 2
const LEADER int = 3

type Raft struct {
	mu        sync.Mutex   // Lock to protect shared access to this peer's state
	peers     []*ClientEnd // RPC end points of all peers
	persister *Persister   // Object to hold this peer's persisted state
	me        int          // this peer's index into peers[]
	dead      int32        // set by Kill()

	applyCh       chan ApplyMsg
	applyStartIdx int
	//Persistent states, update before e responding to RPCs
	currentTerm int
	votedFor    int
	log         []Entry

	//Volatile states
	commitIndex int
	lastApplied int

	//Volatile states used by leader
	nextIndex  []int
	matchIndex []int

	//role infos
	Role int

	electionTimeStamp int64
	electionTimeout   int64

	heartbeatTimeStamp int64
}

type RequestVoteArgs struct {
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

type RequestVoteReply struct {
	Term        int
	VoteGranted bool
}

type AppendEntriesArgs struct {
	Term         int
	LeaderId     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []Entry
	LeaderCommit int
}

type AppendEntriesReply struct {
	Term          int
	Success       bool
	ConflictTerm  int
	ConflictIndex int
}

type RaftState struct {
	CurrentTerm int
	VoteFor     int
	Log         []Entry
}

type electionCounter struct {
	success_count int32
	all_count     int32
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	var term int
	var isleader bool
	term = rf.currentTerm
	isleader = (rf.Role == LEADER)
	return term, isleader
}

func (rf *Raft) roleName() string {
	if rf.Role == LEADER {
		return "leader"
	} else if rf.Role == CANDIDATE {
		return "candidate"
	} else {
		return "follower"
	}
}

func (rf *Raft) persist() {

	w := new(bytes.Buffer)
	e := NewEncoder(w)
	state := RaftState{rf.currentTerm, rf.votedFor, rf.log}
	e.Encode(state)
	rf.persister.SaveRaftState(w.Bytes())
}

func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}

	r := bytes.NewBuffer(data)
	d := NewDecoder(r)
	var state RaftState
	if d.Decode(&state) != nil {
		DPrintf("node[%d]: Decode failed, can not read from Persist", rf.me)
	} else {
		rf.votedFor = state.VoteFor
		rf.currentTerm = state.CurrentTerm
		rf.log = state.Log
	}
}

// must be used within critical section
// index and term of the last entry of log
func (rf *Raft) LastestEntryInfo() (int, int) {
	ThisLastIndex := len(rf.log) - 1
	ThisLastTerm := rf.log[ThisLastIndex].Term
	return ThisLastIndex, ThisLastTerm
}

//now in ms
func (rf *Raft) GetNow() int64 {
	return time.Now().UnixNano() / int64(time.Millisecond)
}

//compare in ms
func (rf *Raft) IsTimeOut(TimeStamp int64, TimeOut int64) bool {
	now := rf.GetNow()
	return (now >= TimeStamp+TimeOut)
}

//all in one, a node can only be one role at a time, so use only one goroutine
func (rf *Raft) timeLoop() {
	for !rf.killed() {
		for rf.Role == FOLLOWER {
			rf.doFollower()
			time.Sleep(time.Duration(10) * time.Millisecond)
		}
		for rf.Role == CANDIDATE {
			rf.doCandidate()
			time.Sleep(time.Duration(10) * time.Millisecond)
		}
		for rf.Role == LEADER {
			rf.doLeader()
			time.Sleep(time.Duration(10) * time.Millisecond)
		}
	}
}

func (rf *Raft) applyToStateMachine(oldCommitIdx int) {
	//commitIndex moved # Rules for Servers 1
	if oldCommitIdx != rf.commitIndex {
		//applied catch up
		if rf.commitIndex > rf.lastApplied {
			for rf.lastApplied < rf.commitIndex {
				rf.lastApplied++
				//TODO apply rf.log[rf.lastApplied]  use goroutine
			}
		}
	}
}

// must be used within critical section
// check if logs is more up-to-date than this raft instance
// what 'up-to-date' means written in Raft Paper 5.4.1
func (rf *Raft) HasUpToDateLogs(LastLogIndex int, LastLogTerm int) bool {
	ThisLastIndex, ThisLastTerm := rf.LastestEntryInfo()
	if LastLogTerm > ThisLastTerm {
		return true
	} else if LastLogTerm < ThisLastTerm {
		return false
	} else {
		return (LastLogIndex >= ThisLastIndex)
	}
}

//单独的goroutine,按顺序推送已经commit的条目
func (rf *Raft) applyChAppend() {
	for !rf.killed() {
		rf.mu.Lock()
		if rf.applyStartIdx < rf.commitIndex {
			rf.applyStartIdx++
			msg := ApplyMsg{true, rf.log[rf.applyStartIdx].Command, rf.applyStartIdx}
			DPrintf("node[%v]: applyChAppend idx %v", rf.me, msg)
			rf.mu.Unlock()
			rf.applyCh <- msg
			continue
		}
		rf.mu.Unlock()
		time.Sleep(time.Duration(10) * time.Millisecond)
	}
}

//立即返回，不需要等
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	if rf.killed() {
		return -1, -1, false
	}
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.Role != LEADER {
		return -1, rf.currentTerm, false
	}
	entry := Entry{command, rf.currentTerm}
	rf.log = append(rf.log, entry)
	rf.persist()
	DPrintf("node[%v]: Start success %v %v %v", rf.me, len(rf.log)-1, rf.currentTerm, command)
	return len(rf.log) - 1, rf.currentTerm, true
}

func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func Make(peers []*ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.Role = FOLLOWER
	rf.currentTerm = 0
	rf.votedFor = -1
	rf.log = make([]Entry, 1)
	rf.log[0] = Entry{nil, 0}
	rf.commitIndex = 0
	rf.lastApplied = 0
	rf.nextIndex = make([]int, len(peers))
	rf.matchIndex = make([]int, len(peers))
	rf.applyCh = applyCh
	rf.applyStartIdx = 0
	rf.electionTimeStamp = rf.GetNow()
	rf.electionTimeout = rand.Int63n(200) + 450
	rf.readPersist(persister.ReadRaftState())
	rf.Convert2Follower(rf.currentTerm)
	go rf.timeLoop()
	go rf.applyChAppend()
	return rf
}
